package com.mall.hadoop.job.abilitystatic;

import java.io.IOException;
import java.util.Map;

import net.sf.json.JSONObject;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 统计八大能力各项分数
 * @author tgy
 *
 */
@SuppressWarnings("unchecked")
public class AbilityStaticMap extends Mapper<LongWritable, Text, Text, IntWritable>{

	public static Logger logger = LoggerFactory.getLogger(AbilityStaticMap.class);
	
	@Override
	protected void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException {
		try{
			Map<String,Object> map = JSONObject.fromObject(value.toString());
			String p = MapUtils.getString(map, "province");
			String c = MapUtils.getString(map, "city");
			String a = MapUtils.getString(map, "area");
			//总分数
			Integer totalScore = MapUtils.getInteger(map, "totalScore");
			//社交能力分数
			Integer social = MapUtils.getInteger(map, "social");
			//艺术能力分数
			Integer art = MapUtils.getInteger(map, "art");
			//语言能力分数
			Integer language = MapUtils.getInteger(map, "language");
			//认知能力分数
			Integer cognition = MapUtils.getInteger(map, "cognition");
			//自理能力分数
			Integer themselves = MapUtils.getInteger(map, "themselves");
			//运动能力分数
			Integer motion = MapUtils.getInteger(map, "motion");
			//精细能力分数
			Integer fine = MapUtils.getInteger(map, "fine");
			//数学能力分数
			Integer mathematics = MapUtils.getInteger(map, "mathematics");
			if(!StringUtils.isEmpty(p)){
				context.write(new Text(p+"sf_sf"), new IntWritable(1));
				context.write(new Text(p+"sf_1"), new IntWritable(totalScore));
				context.write(new Text(p+"sf_2"), new IntWritable(social));
				context.write(new Text(p+"sf_3"), new IntWritable(art));
				context.write(new Text(p+"sf_4"), new IntWritable(language));
				context.write(new Text(p+"sf_5"), new IntWritable(cognition));
				context.write(new Text(p+"sf_6"), new IntWritable(themselves));
				context.write(new Text(p+"sf_7"), new IntWritable(motion));
				context.write(new Text(p+"sf_8"), new IntWritable(fine));
				context.write(new Text(p+"sf_9"), new IntWritable(mathematics));
			}
			//计算市下的数量，求平均分需要
			if(!StringUtils.isEmpty(c)){
				context.write(new Text(p+c+"sh_sh"), new IntWritable(1));
				context.write(new Text(p+c+"sh_1"), new IntWritable(totalScore));
				context.write(new Text(p+c+"sh_2"), new IntWritable(social));
				context.write(new Text(p+c+"sh_3"), new IntWritable(art));
				context.write(new Text(p+c+"sh_4"), new IntWritable(language));
				context.write(new Text(p+c+"sh_5"), new IntWritable(cognition));
				context.write(new Text(p+c+"sh_6"), new IntWritable(themselves));
				context.write(new Text(p+c+"sh_7"), new IntWritable(motion));
				context.write(new Text(p+c+"sh_8"), new IntWritable(fine));
				context.write(new Text(p+c+"sh_9"), new IntWritable(mathematics));
			}
			//计算区下的数量，求平均分需要
			if(!StringUtils.isEmpty(a)){
				context.write(new Text(p+c+a+"sq_dq"), new IntWritable(1));
				context.write(new Text(p+c+a+"sq_1"), new IntWritable(totalScore));
				context.write(new Text(p+c+a+"sq_2"), new IntWritable(social));
				context.write(new Text(p+c+a+"sq_3"), new IntWritable(art));
				context.write(new Text(p+c+a+"sq_4"), new IntWritable(language));
				context.write(new Text(p+c+a+"sq_5"), new IntWritable(cognition));
				context.write(new Text(p+c+a+"sq_6"), new IntWritable(themselves));
				context.write(new Text(p+c+a+"sq_7"), new IntWritable(motion));
				context.write(new Text(p+c+a+"sq_8"), new IntWritable(fine));
				context.write(new Text(p+c+a+"sq_9"), new IntWritable(mathematics));
			}
		}catch(Exception e){
			logger.info("解析八大能力信息出现异常："+e.toString());
		}
	}	
}




